package com.centnet.base.bigdata.hbase;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.regex.Pattern;

/**
 * hbase数据对象工具类
 * Created by veblen on 2017/3/16.
 */
public class HbaseDataUtil {

    protected static Logger logger = LoggerFactory.getLogger(HbaseDataUtil.class);

    /**
     * hbase的分隔符
     */
    public static final String HBASE_SEPARATOR = "|";
    /**
     * 如果数据中包含分隔符的字符串，用该字符串替换。
     */
    public static final String HBASE_SEPARATOR_PLACE = "-";

    /**
     * 字符串split的分隔
     */
    public static final String SPLIT_REGX = Pattern.quote(HBASE_SEPARATOR);

    /**
     * 对象转成hbase的字符串
     *
     * @return
     */
    public static String object2dataStr(String tmp, Object o) {
        String[] split = tmp.split(SPLIT_REGX);
        String result = HBASE_SEPARATOR + tmp + HBASE_SEPARATOR;
        String splitRegx;
        for (int i = 0; i < split.length; i++) {
            String fieldName = split[i];
            try {
                String mappedProperty = String.valueOf(PropertyUtils.getProperty(o, fieldName));
                if (StringUtils.isBlank(mappedProperty) || "null".equals(mappedProperty)) {
                    mappedProperty = "";
                } else {
                    //替换对象树中中的|字符串。
                    mappedProperty = mappedProperty.replaceAll(SPLIT_REGX, HBASE_SEPARATOR_PLACE);
                }

                splitRegx = HBASE_SEPARATOR + fieldName + HBASE_SEPARATOR;

                //给属性名称两边加上|,解决出现b属性名称包含a属性名称情况。
                if (i == 0) {
                    mappedProperty = mappedProperty + HBASE_SEPARATOR;
                } else if (i == split.length - 1) {
                    mappedProperty = HBASE_SEPARATOR + mappedProperty;
                } else {
                    mappedProperty = HBASE_SEPARATOR + mappedProperty + HBASE_SEPARATOR;
                }

                result = result.replace(splitRegx, mappedProperty);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

        //判断数据能否被正确解析
        String[] valArr = StringUtils.splitPreserveAllTokens(result, SPLIT_REGX);
        if (split.length != valArr.length) {
            logger.error("object2dataStr fail: result string is[" + result + "],object class is:[" + o.getClass() + "]");
        }

        return result;
    }


    /**
     * hbase对象转字符串
     *
     * @param str 数据字符串
     * @param tmp 模板字符串
     * @param cls 对象的class
     * @return
     */
    public static <T> T str2Object(String str, String tmp, Class<T> cls) {
        if (str == null) {
            return null;
        }
        try {
            T t = cls.newInstance();
            String[] tmpArr = tmp.split(SPLIT_REGX);
            String[] valArr = StringUtils.splitPreserveAllTokens(str, SPLIT_REGX);
            for (int i = 0; i < valArr.length; i++) {
                String val = valArr[i];
                String name = tmpArr[i];
                BeanUtils.setProperty(t, name, val);
            }
            return t;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 获取String类型的值
     *
     * @param cell
     * @return
     */
    public static String getStringValueFromCell(Cell cell) {
        int valueOffset = cell.getValueOffset();
        int valueLength = cell.getValueLength();
        String value = Bytes.toString(cell.getValueArray(), valueOffset, valueLength);
        return value;
    }
}
